/**事务消息中的事务逻辑写在此处，它会有三种返回结果：
 * 1.LocalTransactionState.COMMIT_MESSAGE:正式发送消息
 * 2.LocalTransactionState.ROLLBACK_MESSAGE：取消消息发送
 * 3.LocalTransactionState.UNKNOW：待定，需回查
 *
 */

import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.concurrent.atomic.AtomicInteger;

public class TransactionListenerImpl implements TransactionListener {
    private AtomicInteger transactionIndex = new AtomicInteger(0);

    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        //本地事务开始执行
        System.out.println("本地事务开始执行.......");

        System.out.println("msg=" + new String(msg.getBody()));
        System.out.println("arg = "+arg);

        //模拟各种可能的返回结果
        int value = transactionIndex.getAndIncrement();
        int status = value % 3;
        switch (status) {
            case 0:
                return LocalTransactionState.UNKNOW;
            case 1:
                return LocalTransactionState.COMMIT_MESSAGE;
            case 2:
                return LocalTransactionState.ROLLBACK_MESSAGE;
            default:
                return LocalTransactionState.COMMIT_MESSAGE;
        }
    }

    //回查接口，当executeLocalTransaction返回LocalTransactionState.UNKNOW时会调用此方法
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        //先根据key去检查本地事务消息是否完成
        String key = msg.getKeys();

        System.out.println("state --" + new String(msg.getBody()));

        //只有如下两种返回值，没必要再返回unknown
        return LocalTransactionState.COMMIT_MESSAGE;   //LocalTransactionState.ROLLBACK_MESSAGE
    }
}
